package team.bluepen.supermarket.web.servlet;

import team.bluepen.supermarket.constant.Constant;
import team.bluepen.supermarket.constant.Error;
import team.bluepen.supermarket.data.dao.GoodRepository;
import team.bluepen.supermarket.data.dao.abstraction.Repository;
import team.bluepen.supermarket.data.entity.Good;
import team.bluepen.supermarket.data.hdfs.HdfsManager;
import team.bluepen.supermarket.data.hdfs.WordCountRunner;
import team.bluepen.supermarket.util.Log;
import team.bluepen.supermarket.util.StringUtil;
import team.bluepen.supermarket.web.base.DataServlet;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * WordCount Servlet
 * @author Kai
 */
@WebServlet(name = "WordCountServlet", urlPatterns = "/api/word-count")
public class WordCountServlet extends DataServlet {
    private static final String TAG = "WordCountServlet";

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        super.doPost(req, resp);

        File dir = new File(Constant.SERVER_TEMP_DIRECTORY);
        if (!dir.mkdirs() && !dir.exists()) {
            sendRedirect(resp, Error.ERROR_IO);
            return;
        }

        File tmpOut = new File(Constant.SERVER_TEMP_DIRECTORY + "/" + StringUtil.getRandomString(16));
        if (!tmpOut.createNewFile()) {
            sendRedirect(resp, Error.ERROR_IO);
            return;
        }

        Repository<Good> goodRepository = new GoodRepository();
        List<Good> goodList = goodRepository.query();
        PrintWriter writer = new PrintWriter(tmpOut);
        int size = goodList.size();
        for (int i = 0; i < size; i++) {
            if (i < size - 1) {
                writer.println(goodList.get(i).getName());
            } else {
                writer.print(goodList.get(i).getName());
            }
            writer.flush();
        }

        writer.close();

        String hdfsInputPath = Constant.TEMP_DIRECTORY + "/" + tmpOut.getName();
        Error upload = HdfsManager.getInstance().upload(tmpOut.getAbsolutePath(), hdfsInputPath);

        Log.i(TAG, "upload to hdfs state: " + upload);


        if (!tmpOut.delete()) {
            sendRedirect(resp, Error.ERROR_IO);
            return;
        }

        HdfsManager.getInstance().mkdirs(Constant.OUTPUT_DIRECTORY);

        String outputPath = Constant.OUTPUT_DIRECTORY + "/" + getTimeFormat() + "-" +StringUtil.getRandomString(6);

        WordCountRunner runner = new WordCountRunner(Constant.HADOOP_URL + "/" + hdfsInputPath,
                Constant.HADOOP_URL + "/" + outputPath);

        boolean s;
        try {
            Log.i(TAG, "Word Count start: input path " + hdfsInputPath + ", output path" + outputPath);
            s = runner.startJob();
        } catch (InterruptedException | ClassNotFoundException e) {
            s = false;
            Log.e(TAG, e);
        }
        HdfsManager.getInstance().delete(hdfsInputPath);
        Error state = s ? Error.SUCCESS : Error.ERROR_IO;
        sendRedirect(resp, state);
    }

    private static void sendRedirect(HttpServletResponse resp, Error error) throws IOException {
        resp.sendRedirect("/file.jsp?error=" + error.getValue());
    }

    private static String getTimeFormat() {
        Date date = new Date();
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
        return format.format(date);
    }

}
